package cn.lagou.spark.work1

import java.util.Properties

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 从kafka读取数据，修改格式后，存入另一个topic
 */
object SparkStreamingRead {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("FileDStream").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))
    // 定义kafka cunsumer相关参数
    val groupId: String = "mygroup1"
    val topics: Array[String] = Array("mytopic1")
    val consumerParams: Map[String, Object] = getKafkaConsumerParameters(groupId)
    // 创建DStream
    val dstream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      // 从kafka中读取数据
      ConsumerStrategies.Subscribe[String, String](topics, consumerParams)
    )
    //便利数据转换后发送到kafka
    dstream.foreachRDD{lineRDD =>
      if(!lineRDD.isEmpty()){
        lineRDD.foreachPartition{iter =>
          iter.map{line =>
            val arr = line.toString.replace("<<<!>>>", "").split(",")
            if (arr.length != 15) return ""
            arr.mkString("|")
          }.filter(!_.isEmpty)
            .foreach{line =>
              val producer = new KafkaProducer[String, String](getKafkaProducerParameters())
              val record = new ProducerRecord[String, String]("mytopic2", line)
              producer.send(record)
            }
        }
      }
    }
    //启动
    ssc.start()
    ssc.awaitTermination()
  }
  def getKafkaConsumerParameters(groupid: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux121:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupid,
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean),
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest"
    )
  }
  def getKafkaProducerParameters(): Properties = {
    val prop = new Properties()
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux121:9092")
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    prop
  }
}
